package com.hhf.rrd.usage.window;

import org.apache.flink.api.common.eventtime.Watermark;
import org.apache.flink.api.common.eventtime.WatermarkGenerator;
import org.apache.flink.api.common.eventtime.WatermarkOutput;
import org.apache.flink.shaded.jackson2.com.fasterxml.jackson.databind.node.ObjectNode;

/**
 * 自定义 watermark
 *
 * @author huanghaifeng15
 * @date 2022/2/15 09:41
 **/
public class PunctuatedAssigner implements WatermarkGenerator<ObjectNode> {

    @Override
    public void onEvent(ObjectNode event, long eventTimestamp, WatermarkOutput output) {
        long time = event.get("time").asLong();
        output.emitWatermark(new Watermark(time));
    }

    @Override
    public void onPeriodicEmit(WatermarkOutput output) {
        // don't need to do anything because we emit in reaction to events above
    }
}
